package com.caissa.springboot.starter.msg.service;

import cn.hutool.core.util.IdUtil;
import com.caissa.retail.msg.export.common.constant.KafkaTopicConstant;
import com.caissa.retail.msg.export.common.enums.MessageTypeEnum;
import com.caissa.retail.msg.export.common.enums.ResendTimeEnum;
import com.caissa.retail.msg.export.po.MsgInfoPo;
import com.caissa.retail.msg.export.po.sms.SmsBatchMsgPo;
import com.caissa.retail.msg.export.po.sms.SmsMsgPo;
import com.base.springboot.starter.kafka.service.KafkaService;
import com.base.springboot.starter.msg.config.ResendPo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.Date;

/**
 * 消息Service
 *
 * @author yuzhongbo
 * @date 2020/5/20
 */
@Slf4j
public class SmsService {

    @Autowired
    private KafkaService kafkaService;

    /**
     * 发送短信消息
     *
     * @param smsMsgPo
     * @return
     */
    public ListenableFuture<SendResult<String, Object>> sendSmsMsg(SmsMsgPo smsMsgPo,  ResendPo resendPo) {
        MsgInfoPo msgInfoPo = new MsgInfoPo();
        msgInfoPo.setId(IdUtil.simpleUUID());
        msgInfoPo.setMessageType(MessageTypeEnum.SMS);
        msgInfoPo.setSendTotalNum(resendPo.getSendTotalNum());
        msgInfoPo.setResendTime(resendPo.getResendTimeEnum());
        msgInfoPo.setSendDateTime(new Date());
        msgInfoPo.setMessageInfo(smsMsgPo);
        log.info("sendSmsMsg:{}-{} ",KafkaTopicConstant.SEND_SMS_TOPIC, msgInfoPo);
        return kafkaService.send(KafkaTopicConstant.SEND_SMS_TOPIC, msgInfoPo);
    }
    /**
     * 批量发送短信消息
     *
     * @param smsBatchMsgPo
     * @return
     */
    public ListenableFuture<SendResult<String, Object>> sendSmsBatchMsg(SmsBatchMsgPo smsBatchMsgPo, ResendPo resendPo) {
        MsgInfoPo msgInfoPo = new MsgInfoPo();
        msgInfoPo.setId(IdUtil.simpleUUID());
        msgInfoPo.setMessageType(MessageTypeEnum.BATCH_SMS);
        msgInfoPo.setSendTotalNum(resendPo.getSendTotalNum());
        msgInfoPo.setResendTime(resendPo.getResendTimeEnum());
        msgInfoPo.setSendDateTime(new Date());
        msgInfoPo.setMessageInfo(smsBatchMsgPo);
        log.info("sendSmsBatchMsg:{}-{} ",KafkaTopicConstant.SEND_SMS_TOPIC, msgInfoPo);
        return kafkaService.send(KafkaTopicConstant.SEND_SMS_TOPIC, msgInfoPo);
    }
}
